package com.siebre.demo.jdbc.queue;

import java.util.Iterator;
import java.util.List;
import java.util.Queue;

import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.RowMapper;

import com.google.common.base.Objects;
import com.google.common.base.Predicate;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.Lists;
import com.google.common.collect.Queues;
import com.google.common.eventbus.EventBus;

/**
 * A read only queue based on JDBC table
 * 
 * @author ZhangChi
 * @since 2015年1月30日
 */
public class JdbcQueue<T> {
	
	private final JdbcOperations jdbcOps;
	
	private final String selectSql;
	
	private final String updateSql;
	
	private final RowMapper<T> rowMapper;
	
	private EventBus eventBus = new EventBus() {

		@Override
		public void post(Object event) {
		}
	};

	JdbcQueue(JdbcOperations jdbcOps, String selectSql, RowMapper<T> rowMapper, String updateSql) {
		this.jdbcOps = jdbcOps;
		this.selectSql = selectSql;
		this.rowMapper = rowMapper;
		this.updateSql = updateSql;
	}
	
	public List<T> poll(int maxItemCount) {
		return pollFunc(maxItemCount);
	}
	
	public List<T> pollFunc(final int maxItemCount) {
		final PollingStats stats = new PollingStats();
		List<T> result = FluentIterable.from(new Iterable<T>() {

			@Override
			public Iterator<T> iterator() {
				return new AbstractIterator<T>() {
					
					private int offset = 0;
					
					private Queue<T> buffer = Queues.newLinkedBlockingQueue();

					@Override
					protected T computeNext() {
						if (buffer.isEmpty()) {
							int limit = maxItemCount * (stats.queryCount + 2);
							//TODO abort query if limit becomes too large
							List<T> queryResult = query(offset, limit);
							stats.queryCount++;
							if (queryResult.isEmpty()) {
								return endOfData();
							}
							buffer.addAll(queryResult);
						}
						return buffer.poll();
					}
				};
			}
		})
		.filter(new Predicate<T>() {

			@Override
			public boolean apply(T input) {
				//TODO transform input to parameter array
				boolean updated = jdbcOps.update(updateSql, input) == 1;
				if (!updated) {
					stats.updateFailedCount++;
				}
				return updated;
			}
		})
		.limit(maxItemCount)
		.toList();
		
		eventBus.post(stats);
		return result;
	}
	
	private List<T> query(int offset, int limit) {
		String fetchFirstRows = selectSql + " offset ? limit ?";
		return jdbcOps.query(fetchFirstRows, new Object[] {offset, limit}, rowMapper);
	}

	public List<T> pollOld(int maxItemCount) {
		List<T> fetched = Lists.newArrayList();
		String fetchFirstRows = selectSql + " offset ? limit ?";
		int offset = 0;
		int limit = maxItemCount;
		
		PollingStats stats = new PollingStats();
		
		while (fetched.size() < maxItemCount) {
			List<T> result = jdbcOps.query(fetchFirstRows, new Object[] {offset, limit}, rowMapper);
			
			if (result.isEmpty()) {
				break;
			}
			
			for (T t : result) {
				boolean updated = jdbcOps.update(updateSql, t) == 1;
				if (updated) {
					fetched.add(t);
					if (fetched.size() >= maxItemCount) {
						break;
					}
				}
			}
			
			offset += limit;
		}

		eventBus.post(stats);
		return fetched;
	}

	public void setEventBus(EventBus eventBus) {
		this.eventBus = eventBus;
	}
	
	/**
	 * Insights of polling process
	 * 
	 * @author ZhangChi
	 * @since 2015年2月1日
	 */
	public static class PollingStats {
		
		private final String threadName;
		
		private int queryCount;
		
		private int updateFailedCount;

		PollingStats() {
			this.threadName = Thread.currentThread().getName();
		}

		@Override
		public String toString() {
			return Objects.toStringHelper(this)
					.add("threadName", threadName)
					.add("queryCount", queryCount)
					.add("updateFailedCount", updateFailedCount)
					.toString();
		}
	}
}
